#include "../TcpConnection.h"
#include "../Channel.h"
#include "../EventLoop.h"
#include "../Socket.h"
namespace yhaida
{
    void defaultConnectionCallback(const TcpConnectionPtr &conn)
    {
        LOG_TRACE << conn->localhostAddress().toHostPort() << " -> "
                  << conn->peerAddress().toHostPort() << " is "
                  << (conn->connected() ? "UP" : "DOWN");
    }

    void defaultMessageCallback(const TcpConnectionPtr &,
                                Buffer *buffer,
                                Timestamp)
    {
        buffer->retrieveAll();
    }
}



using namespace yhaida;

TcpConnection::TcpConnection(EventLoop *loop,
                             const std::string &name,
                             int sockfd,
                             const InetAddress &localAddr,
                             const InetAddress &peerAddr)
    : _mainEventLoop(CHECK_NOTNULL(loop)),
      _name(name),
      _state(StateE::kConnecting),
      _sockfd(new Socket(sockfd)),
      _channel(new Channel(loop, sockfd)),
      _localhostAddress(localAddr),
      _peerAddress(peerAddr),
      _reading(true)
{
    LOG_DEBUG << "TcpConnection::ctor[" << _name << "] at " << this << " fd=" << sockfd;
    //用户注册
    _channel->setReadCallBackFuc(std::bind(&TcpConnection::handleRead, this, _1));
    _channel->setWriteCallBackFuc(std::bind(&TcpConnection::handleWrite, this));
    //内部自定义
    _channel->setErrorCallBackFuc(std::bind(&TcpConnection::handleError, this));
    _channel->setCloseCallBackFuc(std::bind(&TcpConnection::handleClose, this));
}
TcpConnection::~TcpConnection()
{
    LOG_DEBUG << "TcpConnection::dtor[" << _name << "] at " << this << " fd=" << _channel->fd();
}
void TcpConnection::connectEstablished()
{
    _mainEventLoop->assertInLoopThread();
    assert(_state == StateE::kConnecting);
    setState(StateE::kConnected);

    _channel->enableReading();

    _connectionCallback(shared_from_this());
}
//这是TcpConnection析构前的最后一步,断开连接的最后一步
void TcpConnection::connectDestroyed()
{
    _mainEventLoop->assertInLoopThread();
    //连接的状态
    // assert(_state == kConnected || _state == kDisconnecting);
    if (_state == kConnected)
    {
        setState(kDisconnected);
        _channel->disenableAll();

        _connectionCallback(shared_from_this());
    }
    // setState(kDisconnected);
    // // handleClose()已经调用过一遍了，再次调用是因为在某些情况下可以不经由handleClose()而直接调用connectDestroyed();
    // _channel->disenableAll();
    // // 这是连接函数，因为连接函数里面写了if...（连接）...else...（断开)...
    // _connectionCallback(shared_from_this());

    // TcpServer通过queueLoop(TcpConnection::connectDestroyed())到IO线程处理
    _mainEventLoop->removeChannel(get_pointer(_channel));
}
void TcpConnection::handleRead(Timestamp receiveTime)
{
    int savedErrno = 0;
    ssize_t n = _inputBuffer.readFD(_channel->fd(), &savedErrno);
    if (n > 0)
    {
        _messageCallback(shared_from_this(), &_inputBuffer, receiveTime);
    }
    //触发了EPOLLIN但是没数据应该关闭
    else if (n == 0)//被动关闭      这个是FIN 挥手的时候
    {
        handleClose();
    }
    else
    {
        errno = savedErrno;
        LOG_SYSERR << "TcpConnection::handleRead";
        handleError();
    }
}
//_outputBuffer里面有数据的时候
void TcpConnection::handleWrite()
{
    _mainEventLoop->assertInLoopThread();
    if (_channel->isWriting())
    {
        ssize_t writale = ::write(_channel->fd(), _outputBuffer.peek(), _outputBuffer.readableBytes());

        if (writale <= 0)
        {
            LOG_SYSERR << "TcpConnection::handleWrite";
            abort(); // FIXME
        }
        else
        {
            _outputBuffer.retrieve(writale);
            if (_outputBuffer.readableBytes() == 0) // finish
            {
                //当数据写完才关闭fd
                _channel->disenableWriting();
                if (_writeCompleteCallback)
                {
                    //延时运行所以用queueInLoop()而不是runInLoop()
                    _mainEventLoop->queueInLoop(
                        std::bind(_writeCompleteCallback, shared_from_this()));
                }
                if (_state == kDisconnecting)
                {
                    shutdownInLoop();
                }
            }
            else
            {
                LOG_TRACE << "I am going to write more data";
            }
        }
    }
    else
    {
        LOG_TRACE << "Connection is down, no more writing";
    }
}

//不开放
void TcpConnection::handleError()
{
    int err = sockets::getSocketError(_channel->fd());
    LOG_ERROR << "TcpConnection::handleError [" << _name
              << "] - SO_ERROR = " << err << " " << strerror_tl(err);
}
void TcpConnection::handleClose()
{
    _mainEventLoop->assertInLoopThread();
    LOG_TRACE << "TcpConnection::handleClose state = " << _state;
    assert(_state == kConnected || _state == kDisconnecting);
    // 把connfd的event变为不可读不可写
    setState(kDisconnected);
    _channel->disenableAll();
    _connectionCallback(shared_from_this());
    
    //绑定的是void TcpServer::removeConnection(const TcpConnectionPtr& conn);
    _closeCallback(shared_from_this());
}

void TcpConnection::send(const char *str, size_t length)
{
    send(std::string(static_cast<const char *>(str), length));
}
void TcpConnection::send(const std::string &str)
{
    if (_state == kConnected)
    {
        if (_mainEventLoop->isInLoopThread())
        {
            sendInLoop(str);
        }
        else
        {
            void (TcpConnection::*fp)(const std::string& str) = &TcpConnection::sendInLoop;
            _mainEventLoop->runInLoop(
                std::bind(fp, this, str));
        }
    }
}
void TcpConnection::send(Buffer* buff)
{
  if (_state == kConnected)
  {
    if (_mainEventLoop->isInLoopThread())
    {
      sendInLoop(buff->peek(), buff->readableBytes());
      buff->retrieveAll();
    }
    else
    {
      void (TcpConnection::*fp)(const std::string& message) = &TcpConnection::sendInLoop;
      _mainEventLoop->runInLoop(
          std::bind(fp,
                    this,     // FIXME
                    //复制了一份值
                    buff->retrieveAllAsString()));
                    //std::forward<string>(message)));
    }
  }
}
/*
sendInLoop()会先尝试直接发送数据，如果一次发送完毕就不会启用WriteCallback;
如果只发送了部分数据，则把剩余的数据放入_outputBuffer,并开始关注writable事件，
以后在handlerWrite()中发送剩余的数据。如果当前outputBuffer.已经有待发送的数据，
那么就不能先尝试发送了，因为这会造成数据乱序。
*/
void TcpConnection::sendInLoop(const std::string &str)
{
    sendInLoop(str.data(), str.size());
}

void TcpConnection::sendInLoop(const void *str, size_t length)
{
    _mainEventLoop->assertInLoopThread();
    ssize_t writable = 0;
    size_t remaining = length;
    bool faultError = false;
    // _channel->isWriting()说明已经准备在准备写了 && 写缓存里面如果有内容 -> 跳过
    if (!_channel->isWriting() && _outputBuffer.readableBytes() == 0)
    {
        writable = ::write(_channel->fd(), str, length);
        if (writable < 0)
        {
            writable = 0;
            // EWOULDBLOCK表示发送时套接字发送缓冲区已满，或接收时套接字接收缓冲区为空。
            if (errno != EWOULDBLOCK)
            {
                LOG_SYSERR << "TcpConnection::sendInLoop";
                if (errno == EPIPE || errno == ECONNRESET) // FIXME: any others?
                {
                    faultError = true;
                }
            }
        }
        else
        {
            remaining = length - writable;
            if (remaining == 0 && _writeCompleteCallback)
            {
                _mainEventLoop->queueInLoop(
                    std::bind(_writeCompleteCallback, shared_from_this()));
            }
        }
    }
    assert(remaining <= length);
    if (!faultError && remaining > 0)
    {
        size_t oldLen = _outputBuffer.readableBytes();
        if (oldLen + remaining >= _highWaterMark && 
            oldLen < _highWaterMark && _highWaterMarkCallback)
        {
            _mainEventLoop->queueInLoop(std::bind(_highWaterMarkCallback, shared_from_this(), oldLen + remaining));
        }
        //把剩余数据存入写缓存
        _outputBuffer.append(static_cast<const char *>(str) + writable, remaining);
        if (!_channel->isWriting())
        {
            //把event置为写标志！
            _channel->enableWriting();
        }
    }
}
void TcpConnection::shutdown()
{
    LOG_TRACE << "TcpConnection::shutdown"<<" fd: "<<_channel->fd()<<" state = " << _state;
    if (_state == kConnected)
    {
        setState(kDisconnecting);
        _mainEventLoop->runInLoop(std::bind(&TcpConnection::shutdownInLoop, this));
    }
}
void TcpConnection::shutdownInLoop()
{
    _mainEventLoop->assertInLoopThread();
    if (!_channel->isWriting())
    {
        _sockfd->shutdownWrite();
    }
}

void TcpConnection::setTcpNoDelay(bool on)
{
    _sockfd->setTcpNoDelay(on);
}

void TcpConnection::setTcpKeepAlive(bool on)
{
    _sockfd->setTcpKeepAlive(on);
}
void TcpConnection::forceCloseWithDelay(double seconds)
{
  if (_state == kConnected || _state == kDisconnecting)
  {
    setState(kDisconnecting);
    _mainEventLoop->runAfter(
        seconds,
        makeWeakCallback(shared_from_this(),
                         &TcpConnection::forceClose));  // not forceCloseInLoop to avoid race condition
  }
}
void TcpConnection::forceClose()
{
  // FIXME: use compare and swap
  if (_state == kConnected || _state == kDisconnecting)
  {
    setState(kDisconnecting);
    _mainEventLoop->queueInLoop(std::bind(&TcpConnection::forceCloseInLoop, shared_from_this()));
  }
}
void TcpConnection::forceCloseInLoop()
{
  _mainEventLoop->assertInLoopThread();
  if (_state == kConnected || _state == kDisconnecting)
  {
    // as if we received 0 byte in handleRead();
    handleClose();
  }
}

//
void TcpConnection::startRead()
{
  _mainEventLoop->runInLoop(std::bind(&TcpConnection::startReadInLoop, this));
}

void TcpConnection::startReadInLoop()
{
  _mainEventLoop->assertInLoopThread();
  if (!_reading || !_channel->isReading())
  {
    _channel->enableReading();
    _reading = true;
  }
}

void TcpConnection::stopRead()
{
  _mainEventLoop->runInLoop(std::bind(&TcpConnection::stopReadInLoop, this));
}

void TcpConnection::stopReadInLoop()
{
  _mainEventLoop->assertInLoopThread();
  if (_reading || _channel->isReading())
  {
    _channel->disableReading();
    _reading = false;
  }
}